package com.ebs.md.fxmonitor.marketmonitor.esper.listener;

/**
 * Perforce Revision:
 *  $File: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/fxmonitor/marketmonitor/esper/listener/AbstractEsperListenerAndPublisher.java $
 *  $Revision: #1 $  $Change: 115094 $  $DateTime: 2012/09/20 16:32:14 $  $Author: Arik.Sher $
 */

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;
import org.springframework.jms.core.JmsOperations;

import com.ebs.md.fxmonitor.marketmonitor.alert.AbstractAlertMessageCreatorBySingleFact;
import com.ebs.md.fxmonitor.marketmonitor.model.AbsSingleFact;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.StatementAwareUpdateListener;
import com.espertech.esper.event.bean.BeanEventBean;
import com.google.protobuf.Message;
import com.google.protobuf.TextFormat;


public abstract class AbstractEsperListenerAndPublisher<GPB extends Message, SF extends AbsSingleFact>
implements StatementAwareUpdateListener
{
	final protected Logger logger = Logger.getLogger(getClass());
	final protected String shortClassName; // used mainly for logging
	
	protected JmsOperations  jmsOperations; // used for publishing the Events
	protected String         jmsEventPublishDestinationName; // publicationDetination
	
    protected boolean        useMarketMonitor;

	protected AtomicLong     eventSequence4log = new AtomicLong(1000000);
    
    public AbstractEsperListenerAndPublisher ()
    {
    	useMarketMonitor = true;
		shortClassName = getClass().getSimpleName();
    }
	
	public void setJmsOperations(JmsOperations jmsOperations) {
		this.jmsOperations = jmsOperations;
	}

	public void setJmsEventPublishDestinationName(
			String jmsEventPublishDestinationName) {
		this.jmsEventPublishDestinationName = jmsEventPublishDestinationName;
	}

	public void setUseMarketMonitor(boolean useMarketMonitor) {
		this.useMarketMonitor = useMarketMonitor;
	}

	@Override
	public void update(EventBean[] newEvents, EventBean[] oldEvents, EPStatement epstatement, EPServiceProvider epserviceprovider) {
		if ( !useMarketMonitor ) {
			return;
		}		
		
		List<SF> singleFactsList = extractSingleFacts(newEvents);
		
		doPublish(singleFactsList);
	}

	/**
	 * publishing 
	 * 
	 * @param singleFactList
	 */
	protected void doPublish(List<SF> singleFactSList) {
		doPublish(getCreator(singleFactSList));
	}
	
	/**
	 * publishing 
	 * 
	 * @param creator specific to be used before sending
	 */
	protected void doPublish(AbstractAlertMessageCreatorBySingleFact<GPB, SF> creator) {
		long eventSeq4log = eventSequence4log.addAndGet(1);
		GPB alertObject = creator.createAlertObject();
		String alertObject4log = TextFormat.shortDebugString(alertObject); 
		
		logger.info(
			String.format("doPublish:: publishing %s event(seq=%d). alert object: '%s'"
				, creator.getEventType()
				, eventSeq4log
				, alertObject4log
			)
		);
		
		// publishing the JMS message
		jmsOperations.send(jmsEventPublishDestinationName, creator);
		
		if ( logger.isDebugEnabled() ) {
			logger.debug(
			String.format("doPublish:: published  %s event(seq=%d). alert object: '%s'"
				, creator.getEventType()
				, eventSeq4log
				, alertObject4log
			)
			);
		}
	}
	
	protected List<SF> extractSingleFacts(EventBean[] eventBeanArr) {
		List<SF> retList = new ArrayList<SF>(eventBeanArr.length);
		int ix=0;
		
		for ( EventBean curr : eventBeanArr ) {
			ix++;
			if ( curr instanceof BeanEventBean ) {
				SF singleFact = (SF)curr.getUnderlying();
				
				retList.add(singleFact);
			} else {
				logger.warn(
					String.format("could not collect item '%d' because not BeanEventBean. content is: %s"
						, ix
						, curr.toString()
					)
				);
			}
		}
		return retList;
	}
	
	protected abstract AbstractAlertMessageCreatorBySingleFact<GPB, SF> getCreator(List<SF> dataItems);
	
}
